package com.niezhipeng.RPC;

import com.niezhipeng.utils.ConnectFactoryUtils;
import com.rabbitmq.client.*;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.UUID;
import java.util.concurrent.*;
// 相当于是生产者
public class RPCClient implements AutoCloseable {

    private final Connection connection;
    private final Channel channel;
    private static final String requestQueueName = "rpc_queue";

    public RPCClient() throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        ConnectFactoryUtils.setFactoryConfig(factory);

        connection = factory.newConnection();
        channel = connection.createChannel();
    }

    public static void main(String[] argv) {
        try (RPCClient fibonacciRpc = new RPCClient()) {
            for (int i = 0; i < 32; i++) {
                String i_str = Integer.toString(i);
                System.out.println(" [x] Requesting fib(" + i_str + ")");
                String response = fibonacciRpc.call(i_str);
                System.out.println(" [.] Got '" + response + "'");
            }
        } catch (IOException | TimeoutException | InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
    }

    public String call(String message) throws IOException, InterruptedException, ExecutionException {
        final String corrId = UUID.randomUUID().toString();

        // 声明一个发送申请用的队列
        String replyQueueName = channel.queueDeclare().getQueue();
        AMQP.BasicProperties props = new AMQP.BasicProperties
                .Builder()
                .correlationId(corrId)    // 相关标识
                .replyTo(replyQueueName)
                .build();

        // 向队列发送消息
        channel.basicPublish("", requestQueueName, props, message.getBytes(StandardCharsets.UTF_8));

        final CompletableFuture<String> response = new CompletableFuture<>();

        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            if (delivery.getProperties().getCorrelationId().equals(corrId)) {  // 消息中的标识是刚刚发送的那条消息的标识
                response.complete(new String(delivery.getBody(), StandardCharsets.UTF_8));
            }
        };
        String ctag = channel.basicConsume(replyQueueName, true, deliverCallback, consumerTag -> { });

        String result = response.get();
        // 取消这个消费者的订阅
        channel.basicCancel(ctag);
        return result;
    }

    public void close() throws IOException {
        connection.close();
    }
}

